package org.ykhl.sdk.appserver;

import org.ykhl.mq.commons.BaseMessage;
import org.ykhl.mq.commons.ConnectionException;
import org.ykhl.mq.commons.pusher.MessagePusher;
import org.ykhl.mq.commons.pusher.PushMessage;
import org.ykhl.mq.commons.pusher.RequestResponse;
import org.ykhl.mq.commons.receiver.MessageConsumeStatus;
import org.ykhl.mq.commons.receiver.MessageListener;
import org.ykhl.mq.commons.receiver.MessageReceiver;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * Created by Zhangkefei on 2016/1/14.
 */
public class MessageQueueKafkaImpl {

    public static final MessageListener NO_OP_LISTENER = new MessageListener() {
        @Override
        public MessageConsumeStatus consume(BaseMessage msg) {
            return MessageConsumeStatus.CONSUME_SUCCESS;
        }
    };

    private final MessagePusher pusher;
    private final MessageReceiver receiver;

    MessageQueueKafkaImpl(MessagePusher pusher, MessageReceiver receiver) {
        this.pusher = pusher;
        this.receiver = receiver;
    }

    /**
     * 设置MessageQueue连接的服务器地址.
     *
     * @param serverAddress
     */
    public void setServerAddress(String serverAddress){
        pusher.setServerAddress(serverAddress);
        receiver.setServerAddress(serverAddress);
    }

    /**
     * 设置是否使用加密及认证模式进行传输。
     *
     * @param useSsl
     */
    public void setUseSsl(Boolean useSsl) {
        pusher.setUseSsl(useSsl);
        receiver.setUseSsl(useSsl);
    }

    /**
     * 查看MessageQueue当前是否正在运行
     *
     * @return true if running, false otherwise
     */
    public boolean isRunning(){
        return pusher.isRunning() && receiver.isRunning();
    }

    /**
     * 用于启动MessageQueue的方法, 开始使用前需要调用
     *
     * @throws ConnectionException
     */
    public void start() throws ConnectionException {
        receiver.start();
        pusher.start();
    }

    /**
     * 用于停止MessageQueue的方法.
     *
     * @throws ConnectionException
     */
    public void stop() throws ConnectionException{
        try {
            pusher.stop();
        } finally {
            receiver.stop();
        }
    }

    /**
     * 设置推送消息的消息监听器.
     *
     * @param listener
     */
    public void setListener(MessageListener listener){
        receiver.setListener(listener);
    }

    /**
     * 向平台发送医疗信息查询请求
     *
     * @param fid 请求的服务ID
     * @param requestBody 请求体字符串
     * @param deviceSign 请求来源的终端设备的特征码
     * @return 用于后续接受响应消息的阻塞队列
     * @throws ConnectionException
     */
    public BlockingQueue<BaseMessage> sendRequest(String fid, String version, String requestBody,
                                                  String deviceSign,String machineID)
            throws ConnectionException {
        RequestResponse rm = new RequestResponse(fid, version, requestBody, deviceSign,machineID);
        return sendRequest(rm);
    }

    /**
     * 向平台发送医疗信息查询请求
     *
     * @param request 用户生成的request实例
     * @return 用于后续接受响应消息的阻塞队列
     * @throws ConnectionException
     */
    public BlockingQueue<BaseMessage> sendRequestAsync(RequestResponse request)
            throws ConnectionException {

        BlockingQueue<BaseMessage> queue = new LinkedBlockingQueue<>();
        MessageReceiver.putQueueCache(request.getRequestId(), queue);
        checkRunning();
//        if (!pusher.sendMessage(request)) {
//            throw new ConnectionException(
//                    "something wrong while sending request, shouldn't be false");
//        }
        pusher.sendMessageAsync(request);//采用异步发送消息

        return queue;
    }

    public BlockingQueue<BaseMessage> sendRequest(RequestResponse request)
            throws ConnectionException {

        BlockingQueue<BaseMessage> queue = new LinkedBlockingQueue<>();
        MessageReceiver.putQueueCache(request.getRequestId(), queue);
        checkRunning();
        if (!pusher.sendMessage(request)) {
            throw new ConnectionException(
                    "something wrong while sending request, shouldn't be false");
        }
        return queue;
    }

    /**
     * 向平台发送推送消息, 不需要返回响应
     *
     * @param msg
     * @return
     * @throws ConnectionException
     */
    public boolean sendMessage(PushMessage msg) throws ConnectionException{
        checkRunning();

        return pusher.sendMessage(msg);
    }

    private void checkRunning() {
        if (isRunning()) return;

        throw new IllegalStateException(
                "MessageQueue in wrong status:" +
                "\n\tpusher running: " + pusher.isRunning() +
                "\n\treceiver running: " + receiver.isRunning());
    }
}
